/*
 * Created by SharpDevelop.
 * User: Administrator
 * Date: 6/8/2009
 * Time: 5:09 PM
 * 
 * To change this template use Tools | Options | Coding | Edit Standard Headers.
 */
 
using System;
using System.Collections.Generic;
using System.Runtime.Serialization;
using System.Threading;

using Communication;
using Communication.Network;

namespace MapSharpLib
{
	public class MRManager
	{
		JobPusher pusher;
		
		Server<Job> resultsReturn;
		List<Job> results;
		string managerIP;
		int managerPort;
		
		volatile int Status;
		
		Dictionary<string, Job> waitingOn;
		List<NodeDescription> nodeStats;
		Dictionary<string, NodeDescription> nodeStatsDic;
		ReaderWriterLockSlim nStatsLock;
		
		public MRManager(int managerPort)
		{
			//HACK: hard coded for testing.
			managerIP = "127.0.0.1";
			
			this.managerPort = managerPort;
			pusher = new JobPusher();
			waitingOn = new Dictionary<string, Job>();
			results = new List<Job>();
			
			nodeStatsDic = new Dictionary<string, NodeDescription>();
			nodeStats = new List<NodeDescription>();
			nStatsLock = new ReaderWriterLockSlim();
			
			resultsReturn = new Server<Job>(managerPort, new MRMActor(waitingOn,nodeStats,nodeStatsDic,nStatsLock
			                                                          ,results,null));
		}
		
		public void Do(Job giantJob, int chunking)
		{
			Status = -1;
			
			string client = managerIP + ":" + managerPort.ToString();
			
			IEnumerator<ISerializable> ie = giantJob.Inputs.GetEnumerator();
			ie.MoveNext();
			bool keepGoing = true;
						
			for(int count = 0; keepGoing; count++)
			{
				List<ISerializable> buff = new List<ISerializable>();
				 
				for(int i=0;i<chunking && keepGoing; i++)
				{
					buff.Add(ie.Current);
					keepGoing = ie.MoveNext();
				}
				
				string jobname = giantJob.JobName + " " + count.ToString();
				Job j = giantJob.Clone();
				j.JobName = jobname;
				j.Client = client;
				j.Inputs = buff;
				Push(j);
			}
			Status = 0;
		}
		
		private void Push(Job j)
		{
			string node = pusher.Push(j);
			
			#region add job to global waitingOn & node's pendingWork lists
			nStatsLock.EnterWriteLock();
				
			try{
				waitingOn.Add(j.JobName, j);
				NodeDescription nd = nodeStatsDic[node];
				List<string> l = new List<string>(nd.PendingWork);
				l.Add(j.JobName);
				NodeDescription newNodeDesc = new NodeDescription(node, l);
				
				nodeStatsDic[node] = newNodeDesc;
				nodeStats.Remove(nd);
				nodeStats.Add(newNodeDesc);
			}
			finally{
				nStatsLock.ExitWriteLock();
			}
			#endregion
		}
		
		public bool isDone()
		{			
			return (Status==0 && waitingOn.Count==0);
		}
		
		public void Join()
		{
			while(!isDone())
			{
				Thread.Sleep(300);				
			}
		}
		
		public void Stop()
		{
			resultsReturn.Stop();
		}
		
		public IList<string> NodesList
		{
			get{
				return pusher.NodesList;
			}
			set{
				updateNodeStats(value);
				pusher.NodesList = value;
			}
		}
		
		private void updateNodeStats(IList<string> nodelist)
		{
			#region add job to global waitingOn & node's pendingWork lists
			nStatsLock.EnterWriteLock();
				
			try{
				foreach(string k in nodeStatsDic.Keys)
				{
					if(!NodesList.Contains(k))
					{
						nodeStats.Remove(nodeStatsDic[k]);
						nodeStatsDic.Remove(k);
					}
				}
				foreach(string n in nodelist)
				{
					if(!nodeStatsDic.ContainsKey(n))
					{
						NodeDescription nd = new NodeDescription(n,null);
						nodeStatsDic.Add(n,nd );
						nodeStats.Add(nd);
					}
				}
			}				
			finally{
				nStatsLock.ExitWriteLock();
			}				
			#endregion
		}
		
		public IList<NodeDescription> NodeStats
			{
				get{
					List<NodeDescription> retVal = null;
					
					nStatsLock.EnterReadLock();
					
					try{
						retVal = new List<NodeDescription>(nodeStats);
					}
					finally{
						nStatsLock.ExitReadLock();
					}
					
					return retVal;

				}
			}
		
		private class JobPusher
		{
			List<string> nodesList;
			ReaderWriterLockSlim nListLock;
			asyncTransfer<Job> aT;
			int index;
			
			public IList<string> NodesList
			{
				get{
					List<string> retVal = null;
					
					nListLock.EnterReadLock();
					
					try{
						retVal = new List<string>(nodesList);
					}				
					finally{
						nListLock.ExitReadLock();
					}
					
					return retVal;
				}
				set
				{
					IList<string> v = value;
					
					nListLock.EnterWriteLock();
					
					try{
						nodesList.Clear();
						nodesList.AddRange(v);
			        }
			        finally{
			            nListLock.ExitWriteLock();
			        }
				}
			}
			
			public JobPusher()
			{
				nodesList = new List<string>();
				nListLock = new ReaderWriterLockSlim();
				index = -1;
				aT = new asyncTransfer<Job>();
			}
			
			public string Push(Job j)
			{
				string node = null;
				#region get the right node to push to
				nListLock.EnterReadLock();
					
				try{
					index = (index+1)%(nodesList.Count);
					node = nodesList[index];
				}				
				finally{
					nListLock.ExitReadLock();
				}
				#endregion
				
				j.setAtt("workNode", node);
				aT.pushObject(node, j);
				
				return node;
			}
		}
		
		public List<Job> CurrentResults
		{
			get{
				List<Job> retVal = null;
				lock(results)
				{
					retVal = new List<Job>(results);
					results.Clear();
				}
				return retVal;
			}
		}
	}
	
	public class NodeDescription
	{
		string ipaddress;
		int port;
		List<string> workingOn;
		
		public NodeDescription(string node, List<string> pendingWork)
		{
			string[] a = node.Split(new char[]{':'});
			ipaddress = a[0]; port = int.Parse(a[1]);
			if(pendingWork!=null)
				workingOn = pendingWork;
			else
				workingOn = new List<string>();
		}
		
		public string IPaddress
		{
			get{ return ipaddress;}
		}
		
		public int Port
		{
			get{ return port;}
		}
		
		public string[] PendingWork
		{
			get{
				return workingOn.ToArray();
			}
		}
		
		public override string ToString()
		{
			string retVal = "Node " + ipaddress + ":" + port.ToString() + "-Working on ";
			foreach(string s in PendingWork)
				retVal+= s+ ";";
			return retVal;
		}
	}
	
	class MRMActor : IActor<Job>
	{
		ObjectPipe<Job> op;
		Dictionary<string, Job> waitOn;
		List<Job> results;
		List<NodeDescription> nStats;
		Dictionary<string, NodeDescription> nStatsDic;
		ReaderWriterLockSlim nStatsLock;

		public MRMActor(Dictionary<string, Job> waitOn,
						List<NodeDescription> nStats,
						Dictionary<string, NodeDescription> nStatsDic,
						ReaderWriterLockSlim nStatsLock,
		                List<Job> results, ObjectPipe<Job> op)
		{
			this.nStats = nStats;
			this.nStatsDic = nStatsDic;
			this.nStatsLock = nStatsLock;
			this.op = op;
			this.waitOn = waitOn;
			this.results = results;
		}
		
		public IActor<Job> newActor(ObjectPipe<Job> op)
		{
			return new MRMActor(waitOn,nStats,nStatsDic, nStatsLock,
								results, op);
		}
		
		public void Act()
		{
			Job j = op.getObject();
			
			nStatsLock.EnterWriteLock();
			try{
				waitOn.Remove(j.JobName);
				string node = j.getAtt("workNode");
				
				NodeDescription nd = nStatsDic[node];
				
				List<string> l = new List<string>(nd.PendingWork);
				l.Remove(j.JobName);
				NodeDescription newNodeDesc = new NodeDescription(node, l);
				
				nStatsDic[node] = newNodeDesc;
				nStats.Remove(nd);
				nStats.Add(newNodeDesc);
			}
			finally{
				nStatsLock.ExitWriteLock();
			}				
			
			lock(results)
			{
				results.Add(j);
			}
		}
	}
}

